package com.heima.kafka.stream;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.Arrays;

/**
 * @author VectorX
 * @version V1.0
 * @description
 * @date 2024-05-22 15:53:35
 */
@Configuration
@Slf4j
public class KafkaStreamHelloListener
{
    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //创建kstream对象，同时指定从那个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");
        stream.flatMapValues((ValueMapper<String, Iterable<String>>) value -> Arrays.asList(value.split(" ")))
              //根据value进行聚合分组
              .groupBy((key, value) -> value)
              //聚合计算时间间隔
              .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
              //求单词的个数
              .count()
              .toStream()
              //处理后的结果转换为string字符串
              .map((key, value) -> {
                  System.out.println("key:" + key + ",value:" + value);
                  return new KeyValue<>(key
                          .key()
                          .toString(), value.toString());
              })
              //发送消息
              .to("itcast-topic-out");
        return stream;
    }
}
